// Copyright 2021 Nitric Pty Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package queue

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log"
	"net/url"
	"time"

	"github.com/Azure/azure-storage-queue-go/azqueue"
	"github.com/Azure/go-autorest/autorest/adal"
	"github.com/Azure/go-autorest/autorest/azure"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/proto"

	"github.com/nitrictech/nitric/cloud/azure/runtime/env"
	azqueueserviceiface "github.com/nitrictech/nitric/cloud/azure/runtime/queue/iface"
	azureutils "github.com/nitrictech/nitric/cloud/azure/runtime/utils"
	grpc_errors "github.com/nitrictech/nitric/core/pkg/grpc/errors"
	"github.com/nitrictech/nitric/core/pkg/logger"

	queuespb "github.com/nitrictech/nitric/core/pkg/proto/queues/v1"
)

// Set to 30 seconds,
const defaultVisibilityTimeout = 30 * time.Second

type AzqueueQueueService struct {
	client azqueueserviceiface.AzqueueServiceUrlIface
}

var _ queuespb.QueuesServer = &AzqueueQueueService{}

// Returns an adapted azqueue MessagesUrl, which is a client for interacting with messages in a specific queue
func (s *AzqueueQueueService) getMessagesUrl(queue string) azqueueserviceiface.AzqueueMessageUrlIface {
	qUrl := s.client.NewQueueURL(queue)
	// Get a new messages URL (used to interact with messages in the queue)
	return qUrl.NewMessageURL()
}

// Returns an adapted azqueue MessageIdUrl, which is a client for interacting with a specific message (task) in a specific queue
func (s *AzqueueQueueService) getMessageIdUrl(queue string, messageId azqueue.MessageID) azqueueserviceiface.AzqueueMessageIdUrlIface {
	mUrl := s.getMessagesUrl(queue)

	return mUrl.NewMessageIDURL(messageId)
}

// type azTask struct {
// 	Id      string
// 	Payload map[string]interface{}
// }

func (s *AzqueueQueueService) send(ctx context.Context, queueName string, req *queuespb.QueueMessage) error {
	newErr := grpc_errors.ErrorsWithScope("AzqueueQueueService.Enqueue")

	messages := s.getMessagesUrl(queueName)

	// Send the tasks to the queue
	if taskBytes, err := proto.Marshal(req); err == nil {
		taskPayload := base64.StdEncoding.EncodeToString(taskBytes)
		if _, err := messages.Enqueue(ctx, taskPayload, 0, 0); err != nil {
			return newErr(
				codes.Internal,
				"error sending task to queue",
				err,
			)
		}
	} else {
		return newErr(
			codes.Internal,
			"error marshalling the task",
			err,
		)
	}

	return nil
}

func (s *AzqueueQueueService) Enqueue(ctx context.Context, req *queuespb.QueueEnqueueRequest) (*queuespb.QueueEnqueueResponse, error) {
	failedRequests := make([]*queuespb.FailedEnqueueMessage, 0)

	for _, r := range req.Messages {
		// t := task
		// Azure Storage Queues don't support batches, so each task must be sent individually.
		err := s.send(ctx, req.QueueName, r)
		if err != nil {
			failedRequests = append(failedRequests, &queuespb.FailedEnqueueMessage{
				Message: r,
				Details: err.Error(),
			})
		}
	}

	return &queuespb.QueueEnqueueResponse{
		FailedMessages: failedRequests,
	}, nil
}

// AzureQueueItemLease - Represents a lease of an Azure Storages Queues item
// Azure requires a combination of their unique reference for a queue item (id) and a pop receipt (lease id)
// to perform operations on the item, such as delete it from the queue.
type AzureQueueItemLease struct {
	// The ID of the queue item
	// note: this is an id generated by Azure, it's not the user provided unique id.
	ID string
	// lease id, a new popReceipt is generated each time an item is dequeued.
	PopReceipt string
}

// String - convert the item lease struct to a string, to be returned as a NitricTask LeaseID
func (l *AzureQueueItemLease) String() (string, error) {
	leaseID, err := json.Marshal(l)
	return string(leaseID), err
}

// leaseFromString - Unmarshal a NitricTask Lease ID (JSON) to an AzureQueueItemLease
func leaseFromString(leaseID string) (*AzureQueueItemLease, error) {
	var lease AzureQueueItemLease
	err := json.Unmarshal([]byte(leaseID), &lease)
	if err != nil {
		return nil, err
	}
	return &lease, nil
}

// Receive - Receives a collection of tasks off a given queue.
func (s *AzqueueQueueService) Dequeue(ctx context.Context, req *queuespb.QueueDequeueRequest) (*queuespb.QueueDequeueResponse, error) {
	newErr := grpc_errors.ErrorsWithScope("AzqueueQueueService.Dequeue")

	messages := s.getMessagesUrl(req.QueueName)

	dequeueResp, err := messages.Dequeue(ctx, req.Depth, defaultVisibilityTimeout)
	if err != nil {
		return nil, newErr(
			codes.Internal,
			"failed to received messages from the queue",
			err,
		)
	}

	if dequeueResp.NumMessages() == 0 {
		return &queuespb.QueueDequeueResponse{
			Messages: []*queuespb.DequeuedMessage{},
		}, nil
	}

	// Convert the Azure Storage Queues messages into Nitric tasks
	tasks := []*queuespb.DequeuedMessage{}
	for i := int32(0); i < dequeueResp.NumMessages(); i++ {
		m := dequeueResp.Message(i)
		var queueMessage queuespb.QueueMessage

		fmt.Printf("deserializing payload: %s", m.Text)

		// bytePayload := []byte(m.Text)
		bytePayload, err := base64.StdEncoding.DecodeString(m.Text)
		if err != nil {
			return nil, newErr(
				codes.Internal,
				"failed to decode queue item payload",
				err,
			)
		}

		err = proto.Unmarshal(bytePayload, &queueMessage)
		if err != nil {
			// This item could have its visibility timeout reset and be requeued.
			// However, that risks the unprocessable items being reprocessed immediately,
			// causing a loop where the receiver frequently attempts to receive the same item.
			logger.Errorf("failed to deserialize queue item payload: %s", err.Error())
			continue
		}

		lease := AzureQueueItemLease{
			ID:         m.ID.String(),
			PopReceipt: m.PopReceipt.String(),
		}
		leaseID, err := lease.String()
		// This should never happen, it's a fatal error
		if err != nil {
			return nil, newErr(
				codes.Internal,
				"failed to construct queue item lease id",
				err,
			)
		}

		tasks = append(tasks, &queuespb.DequeuedMessage{
			LeaseId: leaseID,
			Message: &queueMessage,
		})
	}

	return &queuespb.QueueDequeueResponse{
		Messages: tasks,
	}, nil
}

// Complete - Completes a previously popped queue item
func (s *AzqueueQueueService) Complete(ctx context.Context, req *queuespb.QueueCompleteRequest) (*queuespb.QueueCompleteResponse, error) {
	newErr := grpc_errors.ErrorsWithScope("AzqueueQueueService.Complete")

	lease, err := leaseFromString(req.LeaseId)
	if err != nil {
		return nil, newErr(
			codes.InvalidArgument,
			"failed to unmarshal lease id value",
			err,
		)
	}

	// Client for the specific message referenced by the lease
	task := s.getMessageIdUrl(req.QueueName, azqueue.MessageID(lease.ID))
	_, err = task.Delete(ctx, azqueue.PopReceipt(lease.PopReceipt))
	if err != nil {
		return nil, newErr(
			codes.Internal,
			"failed to complete task",
			err,
		)
	}

	return &queuespb.QueueCompleteResponse{}, nil
}

const expiryBuffer = 2 * time.Minute

func tokenRefresherFromSpt(spt *adal.ServicePrincipalToken) azqueue.TokenRefresher {
	return func(credential azqueue.TokenCredential) time.Duration {
		if err := spt.Refresh(); err != nil {
			log.Default().Println("Error refreshing token: ", err)
		} else {
			tkn := spt.Token()
			credential.SetToken(tkn.AccessToken)

			return tkn.Expires().Sub(time.Now().Add(expiryBuffer))
		}

		// Mark the token as already expired
		return time.Duration(0)
	}
}

// New - Constructs a new Azure Storage Queues client with defaults
func New() (*AzqueueQueueService, error) {
	queueUrl := env.AZURE_STORAGE_QUEUE_ENDPOINT.String()
	if queueUrl == "" {
		return nil, fmt.Errorf("failed to determine Azure Storage Queue endpoint, environment variable %s not set", azureutils.AZURE_STORAGE_QUEUE_ENDPOINT)
	}

	spt, err := azureutils.GetServicePrincipalToken(azure.PublicCloud.ResourceIdentifiers.Storage)
	if err != nil {
		return nil, err
	}

	cTkn := azqueue.NewTokenCredential(spt.Token().AccessToken, tokenRefresherFromSpt(spt))

	var accountURL *url.URL
	if accountURL, err = url.Parse(queueUrl); err != nil {
		return nil, err
	}

	pipeline := azqueue.NewPipeline(cTkn, azqueue.PipelineOptions{})
	client := azqueue.NewServiceURL(*accountURL, pipeline)

	return &AzqueueQueueService{
		client: azqueueserviceiface.AdaptServiceUrl(client),
	}, nil
}

func NewWithClient(client azqueueserviceiface.AzqueueServiceUrlIface) *AzqueueQueueService {
	return &AzqueueQueueService{
		client: client,
	}
}
